package cn.itcast.function;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.StockIncrBean;
import cn.itcast.constant.Constant;
import cn.itcast.util.DateUtil;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.math.RoundingMode;

/**
 *
 * 个股涨跌幅需要计算得实时指标包含:涨跌、涨跌幅、振幅;具体计算公式如下:
 *      今日涨跌=当前价-前收盘价
 *      今日涨跌幅（%）=（当前价-前收盘价）/ 前收盘价 * 100%
 *      今日振幅 =（当日最高点的价格－当日最低点的价格）/昨天收盘价×100%
 */
public class MinStockIncreWindowFunction extends RichWindowFunction<CleanBean, StockIncrBean,String, TimeWindow> {

    @Override
    public void apply(String s, TimeWindow timeWindow, Iterable<CleanBean> iterable, Collector<StockIncrBean> collector) throws Exception {

        /**
         * 1.新建MinStockIncrWindowFunction 窗口函数
         * 2.记录最新个股
         * 3.格式化日期
         * 4.指标计算
         *   涨跌、涨跌幅、振幅
         * 5.封装输出数据
         */
        //todo 1.新建MinStockIncrWindowFunction 窗口函数

        //todo 2.记录最新个股
        CleanBean cleanBean=null;
        for (CleanBean bean : iterable) {
            if(cleanBean==null){
                cleanBean=bean;
            }
            if(bean.getEventTime()>cleanBean.getEventTime()){
                cleanBean=bean;
            }
        }
        //todo 3.格式化日期
        Long tradeTime = DateUtil.longTimeTransfer(cleanBean.getEventTime(), Constant.format_YYYYMMDDHHMMSS);
        //todo 4.指标计算
        //  涨跌、涨跌幅、振幅
        //今日涨跌=当前价-前收盘价
        //今日涨跌幅（%）=（当前价-前收盘价）/ 前收盘价 * 100%
        //今日振幅 =（当日最高点的价格－当日最低点的价格）/昨天收盘价×100%
        BigDecimal updown = cleanBean.getTradePrice().subtract(cleanBean.getPreClosePrice());
        //保留两位小数，四舍五入
        BigDecimal  increase = updown.divide(cleanBean.getPreClosePrice(),2, RoundingMode.HALF_UP);
        BigDecimal amplitude = (cleanBean.getMaxPrice().subtract(cleanBean.getMinPrice())).divide(cleanBean.getPreClosePrice(),2,RoundingMode.HALF_UP);

        //todo 5.封装输出数据
        StockIncrBean stockIncrBean = new StockIncrBean();
        stockIncrBean.setEventTime(cleanBean.getEventTime());
        stockIncrBean.setSecCode(cleanBean.getSecCode());
        stockIncrBean.setSecName(cleanBean.getSecName());
        stockIncrBean.setIncrease(increase);
        stockIncrBean.setTradePrice(cleanBean.getTradePrice());
        stockIncrBean.setUpdown(updown);
        stockIncrBean.setTradeVol(cleanBean.getTradeVolumn());
        stockIncrBean.setAmplitude(amplitude);
        stockIncrBean.setPreClosePrice(cleanBean.getPreClosePrice());
        stockIncrBean.setTradeAmt(cleanBean.getTradeAmt());
        stockIncrBean.setTradeTime(tradeTime);
        stockIncrBean.setSource(cleanBean.getSource());

        collector.collect(stockIncrBean);
    }
}
